package com.chukun.reactor.master;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.ServerSocketChannel;
import java.nio.channels.SocketChannel;
import java.util.concurrent.Executor;
import java.util.concurrent.Executors;

/**
 * @author chukun@meituan.com
 * @version 1.0.0
 * @description acceptor处理逻辑
 * @create 2022/1/25 21:29
 **/
public class Acceptor implements Runnable {


    /**
     * 计数器
     */
    private int handleNext = 0;

    /**
     * 定义选择器
     */
    private final Selector selector;

    /**
     * 定义ServerSocketChannel
     */
    private final ServerSocketChannel serverSocket;

    // 定义线程数量
    private final int POOL_SIZE = Runtime.getRuntime().availableProcessors();

    // 子线程池，主要用于处理业务逻辑
    private Executor subReactorExecutor = Executors.newFixedThreadPool(POOL_SIZE);

    // 定义子reactor的数量
    private Reactor[] subReactors = new Reactor[POOL_SIZE];

    public Acceptor(Selector selector, int port) throws IOException {
        this.selector = selector;
        serverSocket = ServerSocketChannel.open();
        serverSocket.bind(new InetSocketAddress(port));
        serverSocket.configureBlocking(false);
        serverSocket.register(this.selector, SelectionKey.OP_CONNECT, this);
        init();
    }

    private void init() throws IOException {
        for (int i = 0; i < subReactors.length; i++) {
            subReactors[i] = new Reactor();
            subReactorExecutor.execute(subReactors[i]);
        }
    }


    @Override
    public void run() {
        try {
            // 接收连接，非阻塞模式下，没有连接直接返回 null
            SocketChannel socket = serverSocket.accept();
            if (socket != null) {
                // 把提示发到界面
                socket.write(ByteBuffer.wrap("Multiply Reactor Pattern Example\r\nreactor> ".getBytes()));
                System.out.println(Thread.currentThread().getName() + ":Main-Reactor-Acceptor: " + socket.socket().getLocalSocketAddress() + " 注册到 subReactor-" + handleNext);
                // 如何解决呢，直接调用 wakeup，有可能还没有注册成功又阻塞了。这是一个多线程同步的问题，可以借助队列进行处理
                Reactor subReactor = subReactors[handleNext];
                subReactor.register(new AsyncHandler(socket));
                if (++handleNext == subReactors.length) {
                    handleNext = 0;
                }
            }
        } catch (Exception ex) {
            ex.printStackTrace();
        }
    }
}
